package com.example.provider;

import com.example.domain.Order;
import com.example.provider.constant.MqConstant;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeOrderlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerOrderly;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.common.message.Message;
import org.junit.jupiter.api.Test;

import java.util.Arrays;
import java.util.List;

/**
 * @description: TODO
 * @author: uu
 * @date: 2023/12/10 15:36
 * @version: 1.0
 */
public class OrderMessage {
    private List<Order> list = Arrays.asList(
            new Order("1", 1, "下单"),
            new Order("1", 1, "付款"),
            new Order("1", 1, "物流"),
            new Order("2", 2, "下单"),
            new Order("2", 2, "付款"),
            new Order("2", 2, "物流")
    );
    @Test
    void producer() throws Exception {
        DefaultMQProducer producer = new DefaultMQProducer("order-group");
        producer.setNamesrvAddr(MqConstant.NAME_SRV_ADDR);
        producer.start();
        list.forEach(order -> {
            Message message = new Message("orderTopic", order.toString().getBytes());
            try {
                producer.send(message, (msgQueue, msg, arg)->{
                    int hashCode = arg.toString().hashCode();
                    int i = hashCode % msgQueue.size();

                    return msgQueue.get(i);
                }, order.getId());
            } catch (Exception e) {
                throw new RuntimeException(e);
            }
        });
        producer.shutdown();
    }

    @Test
    void consumer() throws Exception {
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("order-order-group");
        consumer.setNamesrvAddr(MqConstant.NAME_SRV_ADDR);
        consumer.subscribe("orderTopic", "*");
        consumer.registerMessageListener((MessageListenerOrderly)(msgList, context) ->{
            System.out.println(new String(msgList.get(0).getBody()));
            System.out.println(context);
            return ConsumeOrderlyStatus.SUCCESS;
        });
        consumer.start();
        System.in.read();
    }
}
